package com.studerb.service;

import java.util.concurrent.ArrayBlockingQueue;
import java.util.concurrent.Callable;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.Future;
import java.util.concurrent.PriorityBlockingQueue;

import org.apache.commons.lang.SystemUtils;
import org.apache.commons.lang.exception.ExceptionUtils;
import org.apache.log4j.Logger;
import org.springframework.beans.factory.annotation.Autowired;

import com.studerb.events.TaskEvent;
import com.studerb.events.TaskServiceEvent;
import com.studerb.events.TaskServiceEvent.EventType;
import com.studerb.events.TaskServiceEvent.Queue;
import com.studerb.nntp.NntpConnection;
import com.studerb.service.interfaces.IConnectionPoolService;
import com.studerb.service.interfaces.IService;
import com.studerb.service.interfaces.ISettingsService;
import com.studerb.service.interfaces.ITaskService;
import com.studerb.tasks.TaskPriorityComparator;
import com.studerb.tasks.interfaces.ITask;
import com.studerb.tasks.interfaces.ITaskListener;

public class NntpTaskService implements ITaskService, IService {
	private final static String QUEUE_SIZE_KEY = "task.queueSize";
	private final static String FIXED_THREAD_POOL_SIZE = "threadPool.fixedSize";
	private final Logger logger = Logger.getLogger(NntpTaskService.class);

	private boolean started = false;
	private PriorityBlockingQueue<ITask> activeQueue;
	private ArrayBlockingQueue<ITask> failedQueue;
	private ArrayBlockingQueue<ITask> scratchQueue;

	protected ExecutorService executorService;
	@Autowired ISettingsService settingsService;
	@Autowired EventService eventService;

	@Autowired protected IConnectionPoolService connectionPoolService;
	TaskServiceEventListener taskServiceEventListner;
	Thread startMonitor;

	protected boolean addFailedTask(ITask task) {
		boolean success = failedQueue.add(task);
		TaskServiceEvent event = new TaskServiceEvent(task, TaskServiceEvent.EventType.ADDED, TaskServiceEvent.Queue.FAILED);
		eventService.publish(event);
		return success;
	}

	@Override public boolean addTask(ITask task) {
		logger.debug("Adding task to Active: " + task.toString());
		try {
			boolean success = activeQueue.add(task);
			if (success) {
				task.addTaskListener(taskServiceEventListner);
				TaskServiceEvent event = new TaskServiceEvent(task, TaskServiceEvent.EventType.ADDED, TaskServiceEvent.Queue.ACTIVE);
				eventService.publish(event);
			}
			return success;
		}
		catch (Exception e) {
			e.printStackTrace();
			return false;
		}

	}

	private void executeTask(ITask task) {
		logger.debug("Executing Task: " + task);
		executorService.execute(task);
	}

	@Override public int getActiveTasksSize() {
		return activeQueue.size();
	}

	@Override public int getFailedTasksSize() {
		return failedQueue.size();
	}

	@Override public int getScratchTasksSize() {
		return scratchQueue.size();
	}

	@Override public boolean isStarted() {
		return started;
	}

	@Override public boolean moveToScratch(ITask task) {
		boolean success = scratchQueue.add(task);
		TaskServiceEvent event = new TaskServiceEvent(task, TaskServiceEvent.EventType.ADDED, TaskServiceEvent.Queue.SCRATCH);
		eventService.publish(event);
		return success;
	}

	private String printTaskQueueCounts() {
		StringBuilder builder = new StringBuilder();
		builder.append("Active Tasks: " + activeQueue.size() + SystemUtils.LINE_SEPARATOR);
		builder.append("Failed Tasks: " + failedQueue.size() + SystemUtils.LINE_SEPARATOR);
		builder.append("Scratch Tasks: " + scratchQueue.size() + SystemUtils.LINE_SEPARATOR);
		return builder.toString();
	}

	@Override public void start() {
		if (started) {
			throw new RuntimeException("TaskService already Started");
		}
		logger.info("[**NntpTaskService Starting**]");

		int QUEUE_SIZE = Integer.parseInt(settingsService.getSetting(QUEUE_SIZE_KEY));
		logger.debug("Initializing queues with size of: " + QUEUE_SIZE);
		int THREAD_POOL_SIZE = Integer.parseInt(settingsService.getSetting(FIXED_THREAD_POOL_SIZE));
		executorService = Executors.newFixedThreadPool(THREAD_POOL_SIZE);
		activeQueue = new PriorityBlockingQueue<ITask>(QUEUE_SIZE, new TaskPriorityComparator());
		scratchQueue = new ArrayBlockingQueue<ITask>(QUEUE_SIZE);
		failedQueue = new ArrayBlockingQueue<ITask>(QUEUE_SIZE);
		taskServiceEventListner = new TaskServiceEventListener();
		startMonitor = new Thread(new StartTaskMonitor());
		startMonitor.start();
		started = true;
	}

	@Override public void stop() {
		if (!started) {
			throw new RuntimeException("TaskService already Stopped");
		}
		logger.info("[**TaskService Stopping**]");
		// pause for a second in case monitor thread needs service
		startMonitor.interrupt();
		logger.info(printTaskQueueCounts());
		try {
			Thread.sleep(1000);
		}
		catch (InterruptedException ignored) {}
		finally {
			started = false;
		}
	}

	@Override public Future submit(Callable command) {
		logger.debug("Executing command: " + command);
		return executorService.submit(command);
	}

	protected ITask takeFirstActiveTask() throws InterruptedException {
		logger.debug("Taking first from activeQueue");
		return activeQueue.take();
	}

	public class StartTaskMonitor implements Runnable {

		private void processTask() {
			ITask currentTask = null;
			NntpConnection conn = null;
			try {
				logger.debug("Waiting for ActiveTask");
				currentTask = takeFirstActiveTask();
				logger.debug("TaskMonitor: got task: " + currentTask);
				if (currentTask.isConnectionNeeded()) {
					logger.debug("connection needed for Task: " + currentTask);
					try {
						logger.debug("Waiting for connectionPool...");
						conn = connectionPoolService.getConnection();
						logger.debug("TaskMonitor: got Connection: " + conn.toString());
						currentTask.setConnection(conn);
					}
					catch (Exception e) {
						logger.error(ExceptionUtils.getRootCauseMessage(e));
						// TODO - move back to active tasks
						return;
					}
				}
				executeTask(currentTask);
			}
			catch (InterruptedException interruptedException) {
				Thread.currentThread().interrupt();
				if (currentTask != null) {
					logger.debug("Interrupted - adding currentTask to failed...");
					addFailedTask(currentTask);
				}
			}
		}

		@Override public void run() {
			logger.debug("TaskStartMonitor Running...");
			while (!Thread.interrupted()) {
				processTask();
			}
		}
	}

	class TaskServiceEventListener implements ITaskListener {

		@Override public void cancelled(TaskEvent event) {
			ITask task = (ITask) event.getSource();
			logger.debug("Task " + task + " + cancelled.");
		}

		@Override public void doInBackground(TaskEvent event) {}

		@Override public void failed(TaskEvent event) {
			ITask task = (ITask) event.getSource();
			logger.debug("Task " + task + " + failed - moving to Failed tasks.");
			addFailedTask(task);
		}

		@Override public void finished(TaskEvent event) {
			ITask task = (ITask) event.getSource();
			logger.debug("Task finished: " + task);
			if (task.isConnectionNeeded()) {
				logger.debug("returning connection to pool");
				NntpConnection conn = task.getConnectionion();
				task.setConnection(null);
				if (conn == null) {
					throw new RuntimeException("Got null task from nntp connection");
				}
				connectionPoolService.returnConnection(conn);
			}

			TaskServiceEvent evt = new TaskServiceEvent(task, EventType.REMOVED, Queue.ACTIVE);
			eventService.publish(evt);
			task.removeTaskListener(this);
		}

		@Override public void interrupted(TaskEvent event) {
			ITask task = (ITask) event.getSource();
			logger.debug("Task interrupted: " + task);
		}

		@Override public void process(TaskEvent event) {}

		@Override public void succeeded(TaskEvent event) {
			ITask task = (ITask) event.getSource();
			logger.debug("Task Succeeded: " + task);
		}
	}
}
